package cn.bearspark.schedule.provider.service.impl;

import cn.bearspark.article.rpc.ArticleRpc;
import cn.bearspark.common.enumeration.ModuleEnum;
import cn.bearspark.file.rpc.FileRpc;
import cn.bearspark.happening.rpc.HappeningRpc;
import cn.bearspark.schedule.provider.dao.mapper.ScheduledTaskMapper;
import cn.bearspark.schedule.provider.dao.po.ScheduledTaskPO;
import cn.bearspark.schedule.provider.service.ScheduleService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author f
 */
@Service
@EnableScheduling
public class ScheduleServiceImpl implements ScheduleService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleServiceImpl.class);

    @DubboReference(check = false)
    private HappeningRpc happeningRpc;
    @DubboReference(check = false)
    private FileRpc fileRpc;
    @DubboReference(check = false)
    private ArticleRpc articleRpc;

    private final ScheduledTaskMapper scheduledTaskMapper;

    public ScheduleServiceImpl(ScheduledTaskMapper scheduledTaskMapper) {
        this.scheduledTaskMapper = scheduledTaskMapper;
    }

    @Override
    public void scheduled(int moduleCode, LocalDateTime deadline, String info) {
        scheduledTaskMapper.insert(moduleCode, deadline, info);
    }

    /**
     * 每分钟，执行到执行时间的定时任务
     * <p>
     * 注意：本方法调用其它模块的方法必须做 幂等性处理，避免重复消费
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void executeExpired() {
        // 已处理的定时任务 id 集合
        List<Long> executedTaskIdList = new ArrayList<>();

        // 批量消费
        scheduledTaskMapper.listExpired().stream().collect(Collectors.groupingBy(
                ScheduledTaskPO::getModuleCode,
                Collectors.mapping(Function.identity(), Collectors.toList())
        )).forEach((moduleCode, scheduledTaskPOList) -> {
            int size = scheduledTaskPOList.size();
            List<Long> taskIdList = new ArrayList<>(size);
            List<String> taskInfoList = new ArrayList<>(size);
            scheduledTaskPOList.forEach(scheduledTaskPO -> {
                taskIdList.add(scheduledTaskPO.getTaskId());
                taskInfoList.add(scheduledTaskPO.getInfo());
            });
            for (int i = 0; i < 3; i++) {
                // 如果不成功，则再重试 2 次
                try {
                    execute(moduleCode, taskInfoList);
                    executedTaskIdList.addAll(taskIdList);
                    break;
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                    // 重试前先休眠 500ms
                    try {
                        TimeUnit.MILLISECONDS.sleep(500);
                    } catch (InterruptedException ex) {
                        LOGGER.error(ex.getMessage(), ex);
                    }
                }
            }
        });

        // 删除已执行过的定时任务
        if (!executedTaskIdList.isEmpty()) {
            scheduledTaskMapper.deleteExecuted(executedTaskIdList);
        }
    }

    // 将定时任务发给具体的模块
    private void execute(Integer moduleCode, List<String> taskInfoList) {
        if (moduleCode == ModuleEnum.HAPPENING.getCode()) {
            happeningRpc.onHappeningDeadline(taskInfoList);
        } else if (moduleCode == ModuleEnum.FILE.getCode()) {
            fileRpc.onFileDeadline(taskInfoList);
        } else if (moduleCode == ModuleEnum.ARTICLE.getCode()) {
            articleRpc.onArticleDeadline(taskInfoList);
        }
    }

}
